package org.dh2580.delay.queue.bean.task;

import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.dh2580.delay.queue.bean.AbstractMsgListener;
import org.dh2580.delay.queue.bean.bean.DelayQueueBean;
import org.dh2580.delay.queue.bean.db.DBUtil;
import org.dh2580.delay.queue.bean.executor.SimpleExecutor;
import org.dh2580.delay.queue.bean.model.DelayMsg;

/**
 * @Author: hao.deng
 * @Date: 2019/5/15-上午10:25
 */
public class SelectNewMsgTask extends SelectMsgTask {

    private static final String QUERY_NEW_SQL_TPL = "SELECT id AS id, route_key AS routeKey, msg AS msg, ttl AS ttl, ack_state AS ackState, create_time AS createTime FROM tb_delay_msg_queue WHERE ttl > ? AND ttl < ? AND ack_state = 0;";

    public SelectNewMsgTask(SimpleExecutor listenerExecutor, SimpleExecutor callbackExecutor, Map<String, List<AbstractMsgListener>> listeners, Integer maxExpiredMinute) {
        super(listenerExecutor, callbackExecutor, listeners, maxExpiredMinute);
    }


    @Override
    public List<DelayMsg> selectDelayMsgs() throws Exception {
        long ttl = System.currentTimeMillis();

        long minTtl = ttl - TimeUnit.MINUTES.toMillis(this.getMaxExpiredMinute());

        return DBUtil.query(QUERY_NEW_SQL_TPL, DelayMsg.class, minTtl, ttl);
    }
}
